package com.atguigu.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/2/5
 *
 *  SQL不支持基于元素个数的窗口。只支持基于时间的。
 *
 *          https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/sql/queries/window-agg/#group-window-aggregation
 */
public class Demo10_GroupWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = "create table t1( id string,ts bigint ,vc int," +
            "                                       pt AS PROCTIME() ," +
            "                                       et AS TO_TIMESTAMP_LTZ(ts, 3) ," +
            "                                       WATERMARK FOR et AS et - INTERVAL '0.001' SECOND ) with(" +
            "                    'connector' = 'filesystem' ," +
            "                    'path' = 'data/sensor.txt' , " +
            "                    'format' = 'csv'  " +
            "                    )";

        tableEnvironment.executeSql(createTableSql);

        String tumbleSql = "select id,sum(vc) sumVC, " +
            "                       TUMBLE_start(et, INTERVAL '5' SECOND) as wstart  ," +
            "                       TUMBLE_end(et, INTERVAL '5' SECOND) as wend  " +
            "                      from t1 " +
            "                     group by id,TUMBLE(et, INTERVAL '5' SECOND)";

        String slideSql = "select id,sum(vc) sumVC, " +
            "                       HOP_start(et, INTERVAL '3' SECOND,INTERVAL '5' SECOND) as wstart  ," +
            "                       HOP_end(et, INTERVAL '3' SECOND,INTERVAL '5' SECOND) as wend  " +
            "                      from t1 " +
            "                     group by id,HOP(et, INTERVAL '3' SECOND,INTERVAL '5' SECOND)";

        String sessionSql = "select id,sum(vc) sumVC, " +
            "                       SESSION_start(et, INTERVAL '3' SECOND) as wstart  ," +
            "                       SESSION_end(et, INTERVAL '3' SECOND) as wend  " +
            "                      from t1 " +
            "                     group by id,SESSION(et, INTERVAL '3' SECOND)";


        /*
                TUMBLE(time_attr, interval): 构造滚动窗口
                HOP(time_attr, interval  slide, interval size)： 构造滑动窗口
                SESSION(time_attr, interval)： 构造会话窗口
         */
        tableEnvironment.sqlQuery(sessionSql)
                        .execute()
                        .print();

    }
}
